package org.lzfto.flink.demo.config.websocket;

import java.io.IOException;
import java.util.Date;
import java.util.concurrent.ConcurrentHashMap;

import org.lzfto.flink.demo.business.flinkTest.test.SimpleTestService;
import org.lzfto.flink.demo.business.flinkTest.entity.TestInfo;
import org.lzfto.flink.demo.utils.SpringUtils;
import org.lzfto.flink.demo.utils.WsUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;

public class MyWebSocketHandler extends TextWebSocketHandler {
	private final Logger LOGGER = LoggerFactory.getLogger(getClass());

    /**
     *会话，key=sessionId 
     */
    private static final ConcurrentHashMap<String, WsSession> sessions = new ConcurrentHashMap<>();
    
    /**
     * 命名会话，key=名称
     */
    private static final ConcurrentHashMap<String, WsSession> NAMED_SESSIONS = new ConcurrentHashMap<>();

    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
    	String sessionId=session.getId();
        sessions.put(sessionId, new WsSession(sessionId,new Date(),session) );
        session.sendMessage(new TextMessage(WsUtil.buildMessage("您已经连接到服务器(会话ID="+sessionId+")")));
        LOGGER.info("建立ws连接，sessionId={}",sessionId);
    }

    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        // 处理客户端消息
    	String sessId=session.getId();
    	WsSession sess=sessions.get(sessId);
    	sess.setConnectTime(new Date());
    	String msg=message.getPayload();
    	if(msg!=null) {
    		//LOGGER.info("来自ws客户端（id={})的心跳",sessId);
    		if ( msg.equals("heartbeat")) {
    			//心跳包，不做处理
                return;
    		}
    		else if (msg.startsWith("setName:")) {
    			Integer testNo=Integer.valueOf(msg.substring(8));
    			SimpleTestService testService=SpringUtils.getBean(SimpleTestService.class);
    			TestInfo info=testService.getTestInfo(testNo);
    			String name=info.websocketClientName();
                sess.setName(name);
                if (!NAMED_SESSIONS.contains(name)) {
                	NAMED_SESSIONS.put(name, sess);
                }
                LOGGER.info("WebSocket客户端{}已经设置名称:{}",sessId,name);
    		}
    	}
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        sessions.remove(session.getId());
        LOGGER.info("断开ws连接，sessionId={}",session.getId());
    }
    
    private static  void sendTo(TextMessage  message,String receiver) {
    	WsSession session=NAMED_SESSIONS.get(receiver);
        if (session==null) {
        	return;
        }
        WebSocketSession sess=session.getSession();
       
        if (session !=null && sess.isOpen()) {
        	 synchronized(sess) {
        		 try {
     				sess.sendMessage(message);
     			} catch (IOException e) {
     				e.printStackTrace();
     			}
             }
        }
    }
    /**
     * 发送消息到websocket客户端（消息会被封装在PubResult中)
     * @param message  原始消息
     * @param  receiver 接收者，不可空
     * @throws IOException
     */
    public static void sendPubResultTo(String message,String receiver) throws IOException {
        TextMessage textMessage = new TextMessage(WsUtil.buildMessage(message));
        sendTo(textMessage,receiver);
    }
    
    /**
     * 发送数据到
     * @param txt  信息
     * @param  receiver 接收者，不可空
     * @throws IOException
     */
    public static void sendTxtTo(String txt,String receiver) throws IOException {
        TextMessage textMessage = new TextMessage(txt);
        sendTo(textMessage,receiver);
    }
    
    /**
     * 广播
     * @param txt
     * @throws IOException
     */
    public static void broadCastTxt(String txt) throws IOException {
        TextMessage textMessage = new TextMessage(txt);
        for (WsSession session : sessions.values()) {
        	WebSocketSession sess=session.getSession();
        	synchronized(sess) {
        		if (sess.isOpen()) {
                	sess.sendMessage(textMessage);
                }
        	}
        }
    }
    
    public static ConcurrentHashMap<String, WsSession> getWsSessions() {
    	return sessions;
    }
}
